package com.wt.demo;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <dependency>
 * <groupId>com.lmax</groupId>
 * <artifactId>disruptor</artifactId>
 * <version>3.4.2</version>
 * </dependency>
 *
 * @author: wangtao
 * @date:10:01 2018/6/6
 * @email:tao8.wang@changhong.com
 */
public class CustomerProducerDisruptor {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService e = Executors.newCachedThreadPool();
        PCDataFactory factory = new PCDataFactory();
        int bufferSize = 1024;
        Disruptor<PCData> disruptor = new Disruptor<PCData>(factory, bufferSize, e, ProducerType.MULTI, new BlockingWaitStrategy());
        disruptor.handleEventsWithWorkerPool(new Customer(), new Customer(), new Customer(), new Customer());
        disruptor.start();

        RingBuffer<PCData> ringBuffer = disruptor.getRingBuffer();
        Producer p = new Producer(ringBuffer);
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (int i = 0; ; i++) {
            System.out.println("will add Data:" + i);
            bb.putInt(0, i);
            p.pushData(bb);
            Thread.sleep(1000);
        }

    }
}


class PCData {
    int value;

    public int getValue() {
        return value;
    }

    public void setValue(int value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "[" + this.value + "]";
    }
}

class PCDataFactory implements EventFactory<PCData> {

    @Override
    public PCData newInstance() {
        return new PCData();
    }
}

class Customer implements WorkHandler<PCData> {

    @Override
    public void onEvent(PCData s) throws Exception {
        System.out.println("Customer:" + Thread.currentThread().getId() + " custom data:" + s);
    }
}

class Producer {

    private final RingBuffer<PCData> ringBuffer;

    public Producer(RingBuffer<PCData> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void pushData(ByteBuffer bb) {
        long sequence = ringBuffer.next();
        try {
            PCData event = ringBuffer.get(sequence);
            event.setValue(bb.getInt(0));
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}